package com.bjbn.listener;

import com.bjbn.service.ActivemqMonitorService;
import com.bjbn.service.KafkaMapSender;
import com.bjbn.util.XmlHandle;
import lombok.extern.slf4j.Slf4j;
import org.apache.activemq.command.ActiveMQMapMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.TextMessage;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * ActiveMQ消息监听器
 *
 * @author bjbn
 */
@Slf4j
@Component
public class ActivemqMessageListener {

    @Autowired
    private ActivemqMonitorService monitorService;

    @Autowired
    private KafkaMapSender kafkaMapSender;

    /**
     * 监听气象数据Topic
     */
    @JmsListener(destination = "csb.ImportMeteorologicalDataInfoSrv.v0")
    public void onMeteorologicalData(Message message) {
        handleMessage("csb.ImportMeteorologicalDataInfoSrv.v0", message);
    }

    /**
     * 监听ADSB数据Topic
     */
    @JmsListener(destination = "csb.ImportADSBDataInfoSrv.v0")
    public void onADSBData(Message message) {
        handleMessage("csb.ImportADSBDataInfoSrv.v0", message);
    }

    /**
     * 监听DF-Stand数据Topic
     */
    @JmsListener(destination = "csb.ImportDfStandDataInfoSrv.v0")
    public void onDfStandData(Message message) {
        handleMessage("csb.ImportDfStandDataInfoSrv.v0", message);
    }

    /**
     * 监听BS-Stand数据Topic
     */
    @JmsListener(destination = "csb.ImportBsStandDataInfoSrv.v0")
    public void onBsStandData(Message message) {
        handleMessage("csb.ImportBsStandDataInfoSrv.v0", message);
    }

    /**
     * 监听飞管车辆台账数据Topic
     */
    @JmsListener(destination = "csb.ImportFzVehicleDataInfoSrv.v0")
    public void onFzVehicleData(Message message) {
        handleMessage("csb.ImportFzVehicleDataInfoSrv.v0", message);
    }

    /**
     * 监听机坪雷达融合信息Topic
     */
    @JmsListener(destination = "csb.ImportApronRadarDataInfoSrv.v0")
    public void onApronRadarData(Message message) {
        handleMessage("csb.ImportApronRadarDataInfoSrv.v0", message);
    }

    /**
     * 监听飞管车辆轨迹数据Topic
     */
    @JmsListener(destination = "csb.ImportFzLocationDataInfoSrv.v0")
    public void onFzLocationData(Message message) {
        handleMessage("csb.ImportFzLocationDataInfoSrv.v0", message);
    }

    /**
     * 监听机场基础信息Topic
     */
    @JmsListener(destination = "csb.ImportBaseAirportDataInfoSrv.v0")
    public void onBaseAirportData(Message message) {
        handleMessage("csb.ImportBaseAirportDataInfoSrv.v0", message);
    }

    /**
     * 监听机位基础信息Topic
     */
    @JmsListener(destination = "csb.ImportBaseAirstandDataInfoSrv.v0")
    public void onBaseAirstandData(Message message) {
        handleMessage("csb.ImportBaseAirstandDataInfoSrv.v0", message);
    }

    /**
     * 监听航班任务基础信息Topic
     */
    @JmsListener(destination = "csb.ImportBaseAirtaskDataInfoSrv.v0")
    public void onBaseAirtaskData(Message message) {
        handleMessage("csb.ImportBaseAirtaskDataInfoSrv.v0", message);
    }

    /**
     * 监听航司基础信息Topic
     */
    @JmsListener(destination = "csb.ImportBaseAirwayDataInfoSrv.v0")
    public void onBaseAirwayData(Message message) {
        handleMessage("csb.ImportBaseAirwayDataInfoSrv.v0", message);
    }

    /**
     * 监听航班航站动态信息Topic
     */
    @JmsListener(destination = "csb.ImportDFltairlineDataInfoSrv.v0")
    public void onDFltairlineData(Message message) {
        handleMessage("csb.ImportDFltairlineDataInfoSrv.v0", message);
    }

    /**
     * 监听航班值机柜台Topic
     */
    @JmsListener(destination = "csb.ImportDFltcounterDataInfoSrv.v0")
    public void onDFltcounterData(Message message) {
        handleMessage("csb.ImportDFltcounterDataInfoSrv.v0", message);
    }

    /**
     * 监听航班登机门Topic
     */
    @JmsListener(destination = "csb.ImportDFltgateDataInfoSrv.v0")
    public void onDFltgateData(Message message) {
        handleMessage("csb.ImportDFltgateDataInfoSrv.v0", message);
    }

    /**
     * 监听航班主表Topic
     */
    @JmsListener(destination = "csb.ImportDFltinfoDataInfoSrv.v0")
    public void onDFltinfoData(Message message) {
        handleMessage("csb.ImportDFltinfoDataInfoSrv.v0", message);
    }

    /**
     * 监听航班行李提取转盘Topic
     */
    @JmsListener(destination = "csb.ImportDFltlugeftchDataInfoSrv.v0")
    public void onDFltlugeftchData(Message message) {
        handleMessage("csb.ImportDFltlugeftchDataInfoSrv.v0", message);
    }

    /**
     * 监听航班行李滑槽口Topic
     */
    @JmsListener(destination = "csb.ImportDFltlugloadDataInfoSrv.v0")
    public void onDFltlugloadData(Message message) {
        handleMessage("csb.ImportDFltlugloadDataInfoSrv.v0", message);
    }

    /**
     * 监听共享航班Topic
     */
    @JmsListener(destination = "csb.ImportDFltshareDataInfoSrv.v0")
    public void onDFltshareData(Message message) {
        handleMessage("csb.ImportDFltshareDataInfoSrv.v0", message);
    }

    /**
     * 监听航班机位Topic
     */
    @JmsListener(destination = "csb.ImportDFltstandDataInfoSrv.v0")
    public void onDFltstandData(Message message) {
        handleMessage("csb.ImportDFltstandDataInfoSrv.v0", message);
    }

    /**
     * 监听航班保障动态节点Topic
     */
    @JmsListener(destination = "csb.ImportDFltSgDynDataInfoSrv.v0")
    public void onDFltSgDynData(Message message) {
        handleMessage("csb.ImportDFltSgDynDataInfoSrv.v0", message);
    }

    /**
     * 监听旅客人数和行李数量Topic
     */
    @JmsListener(destination = "csb.ImportFlightPassengerAndLuggageDataInfoSrv.v0")
    public void onFlightPassengerAndLuggageData(Message message) {
        handleMessage("csb.ImportFlightPassengerAndLuggageDataInfoSrv.v0", message);
    }

    /**
     * 监听廊桥信息Topic
     */
    @JmsListener(destination = "csb.ImportBridgeDataInfoSrv.v0")
    public void onBridgeData(Message message) {
        handleMessage("csb.ImportBridgeDataInfoSrv.v0", message);
    }

    /**
     * 监听廊桥飞机空调Topic
     */
    @JmsListener(destination = "csb.ImportPlaneAirconditionDataInfoSrv.v0")
    public void onPlaneAirconditionData(Message message) {
        handleMessage("csb.ImportPlaneAirconditionDataInfoSrv.v0", message);
    }

    /**
     * 监听廊桥电源Topic
     */
    @JmsListener(destination = "csb.ImportPowerInfoDataInfoSrv.v0")
    public void onPowerInfoData(Message message) {
        handleMessage("csb.ImportPowerInfoDataInfoSrv.v0", message);
    }

    /**
     * 监听航班季度计划Topic
     */
    @JmsListener(destination = "csb.ImportScheduleFlightDataInfoSrv.v0")
    public void onScheduleFlightData(Message message) {
        handleMessage("csb.ImportScheduleFlightDataInfoSrv.v0", message);
    }

    /**
     * 监听航班临时计划Topic
     */
    @JmsListener(destination = "csb.ImportTemporaryFlightDataInfoSrv.v0")
    public void onTemporaryFlightData(Message message) {
        handleMessage("csb.ImportTemporaryFlightDataInfoSrv.v0", message);
    }

    /**
     * 监听METAR气象报文解析数据Topic
     */
    @JmsListener(destination = "csb.ImportMetarParseDataInfoSrv.v0")
    public void onMetarParseData(Message message) {
        handleMessage("csb.ImportMetarParseDataInfoSrv.v0", message);
    }

    /**
     * 监听TAF气象报文解析数据Topic
     */
    @JmsListener(destination = "csb.ImportTafParseDataInfoSrv.v0")
    public void onTafParseData(Message message) {
        handleMessage("csb.ImportTafParseDataInfoSrv.v0", message);
    }


    /**
     * 处理消息的通用方法
     */
    private void handleMessage(String topicName, Message message) {
        try {
            String messageText = "";
            if (message instanceof TextMessage) {
                messageText = ((TextMessage) message).getText();
            }

            log.debug("收到Topic {} 的消息: {}", topicName, messageText);

            // 更新监控服务中的最新数据时间
            monitorService.updateTopicLastDataTime(topicName);

        } catch (JMSException e) {
            log.error("处理Topic {} 消息时发生错误", topicName, e);
        }
    }

    /**
     * 处理消息的通用方法
     */
    private void handleMessageNew(String topicName, Message message) {
        try {
            String messageText = "";
            if (message instanceof ActiveMQMapMessage) {
                ActiveMQMapMessage message1 = (ActiveMQMapMessage) message;
                Map map = message1.getContentMap();
                String payload = map.get("payload").toString();
                List<Map<String, String>> lst = new ArrayList<>();
                String topic = "";
                // 解析消息内容
                switch (topicName) {
                    case "csb.ImportMeteorologicalDataInfoSrv.v0":
                        lst = XmlHandle.handleMeteorologicalData(payload);
                        topic = "csb_meteorological_out";
                        break;
                    case "csb.ImportADSBDataInfoSrv.v0":
                        lst = XmlHandle.handleADSBData(payload);
                        topic = "aircraft_adsb_data_out";
                        break;
                    case "csb.ImportDfStandDataInfoSrv.v0":
                        lst = XmlHandle.handleDfStandData(payload);
                        topic = "csb_df_stand_out";
                        break;
                    case "csb.ImportBsStandDataInfoSrv.v0":
                        lst = XmlHandle.handleBsStandData(payload);
                        topic = "csb_bs_stand_out";
                        break;
                    case "csb.ImportFzVehicleDataInfoSrv.v0":
                        lst = XmlHandle.handleFzVehicleData(payload);
                        topic = "csb_fz_vehicle_out";
                        break;
                    case "csb.ImportApronRadarDataInfoSrv.v0":
                        lst = XmlHandle.handleApronRadarData(payload);
                        topic = "csb_apron_radar_out";
                        break;
                    case "csb.ImportFzLocationDataInfoSrv.v0":
                        lst = XmlHandle.handleFzLocationData(payload);
                        topic = "csb_fz_location_data_out";
                        break;
                    case "csb.ImportBaseAirportDataInfoSrv.v0":
                        lst = XmlHandle.handleBaseAirportData(payload);
                        topic = "csb_imf_base_airport_out";
                        break;
                    case "csb.ImportBaseAirstandDataInfoSrv.v0":
                        lst = XmlHandle.handleBaseAirstandData(payload);
                        topic = "csb_imf_base_airstand_out";
                        break;
                    case "csb.ImportBaseAirtaskDataInfoSrv.v0":
                        lst = XmlHandle.handleBaseAirtaskData(payload);
                        topic = "csb_imf_base_airtask_out";
                        break;
                    case "csb.ImportBaseAirwayDataInfoSrv.v0":
                        lst = XmlHandle.handleBaseAirwayData(payload);
                        topic = "csb_imf_base_airway_out";
                        break;
                    case "csb.ImportDFltairlineDataInfoSrv.v0":
                        lst = XmlHandle.handleDFltairlineData(payload);
                        topic = "csb_imf_d_fltairline_out";
                        break;
                    case "csb.ImportDFltcounterDataInfoSrv.v0":
                        lst = XmlHandle.handleDFltcounterData(payload);
                        topic = "csb_imf_d_fltcounter_out";
                        break;
                    case "csb.ImportDFltgateDataInfoSrv.v0":
                        lst = XmlHandle.handleDFltgateData(payload);
                        topic = "csb_imf_d_fltgate_out";
                        break;
                    case "csb.ImportDFltinfoDataInfoSrv.v0":
                        lst = XmlHandle.handleDFltinfoData(payload);
                        topic = "csb_imf_d_fltinfo_out";
                        break;
                    case "csb.ImportDFltlugeftchDataInfoSrv.v0":
                        lst = XmlHandle.handleDFltlugeftchData(payload);
                        topic = "csb_imf_d_fltlugeftch_out";
                        break;
                    case "csb.ImportDFltlugloadDataInfoSrv.v0":
                        lst = XmlHandle.handleDFltlugloadData(payload);
                        topic = "csb_imf_d_fltlugload_out";
                        break;
                    case "csb.ImportDFltshareDataInfoSrv.v0":
                        lst = XmlHandle.handleDFltshareData(payload);
                        topic = "csb_imf_d_fltshare_out";
                        break;
                    case "csb.ImportDFltstandDataInfoSrv.v0":
                        lst = XmlHandle.handleDFltstandData(payload);
                        topic = "csb_imf_d_fltstand_out";
                        break;
                    case "csb.ImportDFltSgDynDataInfoSrv.v0":
                        lst = XmlHandle.handleDFltSgDynData(payload);
                        topic = "csb_imf_d_flt_sg_dyn_out";
                        break;
                    case "csb.ImportFlightPassengerAndLuggageDataInfoSrv.v0":
                        lst = XmlHandle.handleFlightPassengerAndLuggageData(payload);
                        topic = "csb_d_flt_passenger_out";
                        break;
                        /*
                    case "csb.ImportBridgeDataInfoSrv.v0":
                        lst = XmlHandle.handleBridgeData(payload);
                        break;
                    case "csb.ImportPlaneAirconditionDataInfoSrv.v0":
                        lst = XmlHandle.handlePlaneAirconditionData(payload);
                        break;
                    case "csb.ImportPowerInfoDataInfoSrv.v0":
                        lst = XmlHandle.handlePowerInfoData(payload);
                        break;

                    case "csb.ImportScheduleFlightDataInfoSrv.v0":
                        lst = XmlHandle.handleScheduleFlightData(payload);
                        break;
                    case "csb.ImportTemporaryFlightDataInfoSrv.v0":
                        lst = XmlHandle.handleTemporaryFlightData(payload);
                        break;
                         */
                }

                for (Map<String, String> item : lst) {
                    try {
                        kafkaMapSender.sendMap(topic, item);
                    } catch (Exception e) {
                        log.error("发送Kafka消息时发生错误", e);
                    }
                }

            }

            log.debug("收到Topic {} 的消息: {}", topicName, messageText);

            // 更新监控服务中的最新数据时间
            monitorService.updateTopicLastDataTime(topicName);

        } catch (JMSException e) {
            log.error("处理Topic {} 消息时发生错误", topicName, e);
        }
    }
} 